package com.bungarus.smart.busi;

import com.bungarus.cluster.ClusterChangeListener;
import com.bungarus.cluster.ClusterInfo;
import com.bungarus.cluster.ClusterManager;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.MessageBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Component
@Qualifier("scanner")
public class Scanner implements ClusterChangeListener {
    final Logger logger = Logger.getLogger(Scanner.class);

    private final static String CLUSTER_TAG_ACCESS = "access";
    /*
     * Tag of target cluster
     */
    private final static String SCAN_CLUSTER_TAG = CLUSTER_TAG_ACCESS;
    /**
     * Target cluster ID, which should be fixed to this smart node
     */
    public final static int SCAN_CLUSTER_ID = 0;

    private ClusterManager clusterManager = new ClusterManager(CLUSTER_TAG_ACCESS);
    /**
     * heartbeat port of target server
     */
    public static final int PORT_7700 = 7700;

    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

    /**
     * Interval of checking, Unit: second
     */
    private final static int PERIOD = 5;
    /**
     * Record how many times retried to ping target server
     */
    private static Map<String, Integer> serverPingRetryTimes = new ConcurrentHashMap<>(0);

    private static Map<String, Channel> serverChannels = new ConcurrentHashMap<>(0);
    private static Map<String, String> serverWANAddresses = new ConcurrentHashMap<>(0);
    protected Map<String, ArrayList<NodeInfo>> remoteServerNodes = new HashMap<>(0);

    private static RecommendStrategy recommendStrategy = new RecommendStrategyNormal();
    private AccessContainer accessContainer = AccessContainer.getInstance();

    private ScheduledExecutorService service = Executors.newScheduledThreadPool(1);

    public Scanner(){
    }

    @PostConstruct
    public void init() throws Exception{
        clusterManager.addClusterChangeListener(this);
        accessContainer.setClusterId(getClusterId());

        service.scheduleAtFixedRate(() -> {
            logger.info("Scanner is running...");
            getNodesInCluster(SCAN_CLUSTER_TAG);
            remoteServerNodes.get(SCAN_CLUSTER_TAG).stream().forEach(node -> {
                logger.info("Checking node " + node.getIp());
                checkNodeConnection(node);
            });
            try {
                logger.debug("Current active access nodes is -> " + accessContainer.getAccessList().toString());
            }
            catch (Exception e) {
                e.printStackTrace();
            }

        }, 0, PERIOD, TimeUnit.SECONDS);
    }

    @PreDestroy
    public void destroy() {
        if(null != service) {
            service.shutdown();
        }
    }

    /**
     * Check node
     * @param node
     */
    protected void checkNodeConnection(NodeInfo node){
        if(!serverChannels.containsKey(node.getIp())){
            connect(node, PORT_7700);
        }
    }

    /**
     * Get newest node list and return the total size
     * @param tag
     * @return  size of the nodes
     */
    protected  void getNodesInCluster(String tag) {
        List<ClusterInfo> clusters;
        try {
            clusters = clusterManager.clusters(tag);
            if (null != clusters && clusters.size() > 0) {
                List<NodeInfo> nodes = clusterManager.nodes(tag, getClusterId());
                remoteServerNodes.put(tag, new ArrayList<>(nodes));
            }
        }
        catch (Exception e) {
            logger.info(e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * Get current cluster id bind to this smart route, which should be certain
     * @return
     */
    public static int getClusterId(){
        // current access cluster id bind to this smart scanner
        return SCAN_CLUSTER_ID;
    }

    private void connect(NodeInfo node, int port) {
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelSmartHandlerInitializer(this, node.getIp()));

            ChannelFuture channelFuture = bootstrap.connect(node.getIp(), port).sync();
            Channel channel = channelFuture.channel();
            channelFuture.addListener(future -> {
                if(future.isSuccess()){
                    addRemoteChannel(node, channel);
                    logger.info("Connected to -> " + node.getIp() + ":" + port);
                } else {
                    logger.error("Connect failed to -> " + node.getIp() + ":" + port);
                }
            });
        } catch (Exception e) {
            logger.error("Exception occurred while connecting to -> " + node.getIp() + ":" + port);
        } finally {
        }
    }


    public void removeRemoteChannel(String host){
        try {
            serverChannels.remove(host);
            serverWANAddresses.remove(host);
            updateAccessList();
        }
        catch (Exception e){
            e.printStackTrace();
        }
    }

    public void addRemoteChannel(NodeInfo node, Channel channel){
        try {
            serverChannels.put(node.getIp(), channel);
            serverWANAddresses.put(node.getIp(), node.getExtraInfo());
            updateAccessList();
        }
        catch(Exception e){
            e.printStackTrace();
        }
    }

    /**
     * Update cache value for access list
     * @throws Exception
     */
    private void updateAccessList() throws Exception{
        accessContainer.saveAccessList(new ArrayList<String>(serverWANAddresses.values()));
    }

    @Override
    public List<String> getClusterTags() {
        return new ArrayList<String>(){{
            add(SCAN_CLUSTER_TAG);
        }};
    }

    @Override
    public void onClusterNodeAdded(NodeInfo node){
        logger.info("Node added, connect the node -> " + node.getIp());
        if(!serverChannels.containsKey(node.getIp())){
            connect(node, PORT_7700);
        }
    }

    @Override
    public void onClusterNodeRemoved(NodeInfo node){
        logger.info("Node removed, delete the channel -> " + node.getIp());
        if(serverChannels.containsKey(node.getIp())){
            removeRemoteChannel(node.getIp());
        }
    }

}
